package com.example.reactstream;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class ReactProcessor extends SubmissionPublisher<String>  implements Flow.Processor<String,String> {

    private Flow.Subscription subscription;

    //发布者跟处理器建立关系
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription=subscription;
        System.out.println("processor 建立订阅关系");
        subscription.request(1);

    }

    @Override
    public void onNext(String s) {
        System.out.println("processor 接收数据："+s);

        //中间处理，处理完再发给订阅者
        this.submit(s.toUpperCase());

        //背压实现的核心
        this.subscription.request(1);

    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("processor 出现异常");

    }

    @Override
    public void onComplete() {
        System.out.println("processor 处理完成数据");

    }
}
